package sparkstreaming.lesson05

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by Administrator on 2018/5/11.
  */
object WordBlack {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetWordCount")
    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc,Seconds(2))

    /**
      * 数据的输入
      */

    /**
      * 自己模拟一个黑名单：
      * 各位注意：
      * 这个黑名单，一般情况下，不是我们自己模拟出来，应该是从mysql数据库
      * 或者是Reids 数据库，或者是HBase数据库里面读取出来的。
      */
    val wordBlackList = ssc.sparkContext.parallelize(List("?", "!", "*"))
      .map(param => (param, true))

    val blackList = wordBlackList.collect()

    val blackListBroadcast = ssc.sparkContext.broadcast(blackList)

    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop1",9999)

    /**
      * 数据的处理
      */

    val wordOneDStream = dstream.flatMap(_.split(","))
      .map((_, 1))
    //transform 需要有返回值，必须类型是RDD
    val wordCountDStream = wordOneDStream.transform(rdd => {

      /**
        * SparkCore:
        *     咱们演示的就是对RDD进行操作
        * SparkSQL:
        *     DataFrame
        *     SQL
        */

      val filterRDD: RDD[(String, Boolean)] = rdd.sparkContext.parallelize(blackListBroadcast.value)
      val resultRDD: RDD[(String, (Int, Option[Boolean]))] = rdd.leftOuterJoin(filterRDD)
      /**
        *
        * (String, (Int, Option[Boolean])
        * String: word
        * Int: 1
        * Option：有可能join上 也有可能join不上
        *
        * 思路：
        * 我们应该要的是join不上的，说白了要的是 Option[Boolean] =None
        *
        * filter:
        * true代表我们要
        */
      resultRDD.filter(tuple => {
        tuple._2._2.isEmpty
      }).map(_._1)

    }).map((_, 1)).reduceByKey(_ + _)
    wordCountDStream

    /**
      * 数据的输出
      *
      */
    wordCountDStream.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
